package operators;

import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import pojo.UniqueView;
import pojo.UserBehavior;

import java.util.HashSet;
import java.util.Set;

public class UV_AllWindowFunction implements AllWindowFunction<UserBehavior, UniqueView, TimeWindow> {
    @Override
    public void apply(TimeWindow window, Iterable<UserBehavior> values, Collector<UniqueView> out) throws Exception {

        Set<Long> userSet = new HashSet<>();
        for (UserBehavior value : values) {
            userSet.add(value.getUserId());
        }

        out.collect(new UniqueView(
                "uv"
                , window.getEnd()
                , userSet.size()
        ));
    }
}
